package com.xzc.wc

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala._

//stream process
object StreamWordCount {
  def main(args: Array[String]): Unit = {
    //create a stream execution environment
    val env = StreamExecutionEnvironment.getExecutionEnvironment
//    env.setParallelism(4)

    //get socket host and port from args
    val paramTool: ParameterTool = ParameterTool.fromArgs(args)
    val host: String = paramTool.get("host")
    val port: Int = paramTool.getInt("port")

    //read data from socket
    val inputDataStream: DataStream[String] = env.socketTextStream(host, port)

    //statistics,participle, group by key:word,aggregation statistics
    val resultDataStream: DataStream[(String, Int)] = inputDataStream
      .flatMap(_.split(" ")) //x => x.split(" ")
      .filter(_.nonEmpty) //default cpu core count
//      .map((_, 1)).setParallelism(3)
      .map((_, 1))
      .keyBy(0) //hash to partition
//      .sum(1).setParallelism(2)
      .sum(1)

    //print
    resultDataStream.print().setParallelism(1) //out put to file ordered,avoid multhreads problem

    //start task execution
    env.execute("stream word count")
  }
}
